
/* copyright (c) 1996, 1997, 1998, 1999 William R. Pearson and the
   U. of Virginia */

/* $Name: fasta34t11d4 $ - $Id: p2_workcomp.c,v 1.14 2002/01/25 04:06:25 wrp Exp $ */

/* This version is modifed to read all files, query and database,
   through the manager process. Workers will now receive their
   database from the manager, rather than reading it themselves.  This
   cuts down considerably on NFS traffic, simplifies searches of
   multiple files, and allows use of clusters of slave nodes that do
   not have NFS access */

/* September, 1994 - this version has been modified to do two kinds of
   searches, a general library search, or list of library sequences search.
   The latter would be used to generate optimized scores for fasta and
   to produce alignments */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#ifdef PVM_SRC
#include "pvm3.h"
#endif

#ifdef MPI_SRC
#include "mpi.h"
#endif

/*
#define PvmDataDefault 0
#define PvmTaskDefault 0
*/
#include "msg.h"
#include "defs.h"
#include "param.h"
#include "w_mw.h"
#include "structs.h"

#ifdef MPI_SRC
#define XTERNAL
#endif
#include "upam.h"
#include "uascii.h"

#ifdef PVM_SRC
int worker, mytid;
int nnodes, pinums[MAXNOD];
#endif
int hosttid;

extern void alloc_pam (int d1, int d2, struct pstruct *ppst); /* allocate ppst->pam12,pam12x */
extern void w_init ();
extern void irand(int);
extern void revcomp(unsigned char *, int, int *);

void init_work (unsigned char *aa0, int n0,
		struct pstruct *ppst, void **f_arg);

void do_work (unsigned char *aa0, int n0, unsigned char *aa1, int n1,
	      int frame,
	      struct pstruct *ppst, void *f_str, struct rstruct *rst);

extern void do_opt (unsigned char *aa0, int n0, unsigned char *aa1, int n1,
		    int frame, struct pstruct *pst, void *f_str,
		    struct rstruct *rst);

extern int calc_id(const unsigned char *aa0, const int n0,
	    const unsigned char *aa1, const int n1,
	    int *res, int nres,
	    struct a_struct *aln, struct pstruct pst,
	    void *f_str);

extern int  do_walign (unsigned char *aa0, int n0, unsigned char *aa1, int n1,
		int frame,
		 struct pstruct *ppst, void *f_str, 
		 int **ares, int *nres, struct a_struct *aln);

extern int calcons(unsigned char *aa0, int n0,
	    unsigned char *aa1, int n1,
	    int *res, int nres, int *nc,
	    struct a_struct *aln, struct pstruct pst,
	    char *seqc0, char *seqc1,void *f_str);

extern void initseq(char **seqc0, char **seqc1, int seqsiz);
extern void freeseq(char **seqc0, char **seqc1);

extern void close_work (unsigned char *aa0, int n0, struct pstruct *ppst,
			void **f_arg);
extern void get_param (struct pstruct *pstr, char *pstring2, char *pstring3);
extern int shuffle(unsigned char *, unsigned char *, int);
extern int wshuffle(unsigned char *, unsigned char *, int, int, int *);
extern int E1_to_s(double eval, int n0, int n1);

extern char err_str[];
void w_abort (p, p1)
char *p, *p1;
{
    fprintf (stderr, " %s %s\n", p, p1);
#ifdef PVM_SRC
    pvm_exit();
    exit (1);
#endif
#ifdef MPI_SRC
  MPI_Abort(MPI_COMM_WORLD,1);
#endif
}

#ifdef PVM_SRC
main ()
#endif
#ifdef MPI_SRC
void
workcomp(int worker)
#endif
{
  unsigned char *aa0[6], *aa1s;	/* Query and library sequences */
  struct mngmsg m_msg;	/* start message from manager to worker 1 */
  struct qmng_str qm_msg; /* updated for each query */
  int last_msg_b[10];	/* last set of numbers */
  struct sqs2 *seqpt; 		/* sequence pointers for chunk */
  int seqbuf_n,seqbuf_s;	/* number of sequences, length of sequences */
  int max_sql;			/* maximum number of sequences/node */
  int *n1_arr, *m_seqnm_arr;
  unsigned char *seq_buf;	/* space for sequence data */
  int ntx;
  long curtype = ONETYPE;	/* current send message type */
  int ieven=0;			/* flag for window shuffle */
  int cur_n0;
  int n1, n1over;		/* length of query, library sequences */
  struct comstr bestr[BFR+1];	/* best structures */
  struct comstr2 bestr2[BFR2+1];	/* best structures */
  struct a_struct *aln_dp;
  int qres_bufsize;		/* results buffer size */
  int bestcnt = 0;		/* how many best structures are full */
  char gstring2[MAX_STR];		/* parameter string for manager */
  char gstring3[MAX_STR];		/* parameter string for manager */
  struct pstruct pst;		/* parameter structure */
  struct rstruct  rst;	/* results structure */
  void *f_str[6];
  int *ares, nres;		/* alignment array, array length */
  struct a_struct aln;		/* alignment boundaries */
  int sw_score;
  int lcnt, count, seqnm;	/* counters */
  int lend;			/*global library sequence number information */
  int lsn;			/* library sequence number */
  struct stage2_str *liblist=NULL;	/* list of sequences to search */
  int i, j;		/* my turn to send sequence descriptions */
  char libstr[21];
  char errstr[128];
  int itt=0;
  int bufid;
  char *seqc0, *seqc1;
  int maxc, lc, nc, nident, ngap;
  float percent, gpercent;

#ifdef MPI_SRC
  MPI_Status mpi_status;
  hosttid = 0;
#endif
  
#ifdef PVM_SRC
  mytid = pvm_mytid();
  hosttid = pvm_parent();
#endif

  w_init();	/* sets up default sascii, hsq, sq */

  /* Allocate space for the query sequence */
  if ((aa0[0] = (unsigned char *) malloc ((MAXTST+2)*sizeof (char))) == NULL) {
    w_abort ("Unable to allocate sequence array[0] - exiting!","");
  }
  *aa0[0]='\0';
  aa0[0]++;

  /* initial messages set up various parameter structures:

     STARTTYPE0: &nnodes
     		 pinums
		 &m_msg

     STARTTYPE1  &pst

     STARTTYPE2	 pam12
     STARTTYPE3	 pam12x
  */

#ifdef PVM_SRC
#ifdef ROUTE_DIRECT
  pvm_setopt(PvmRoute,PvmRouteDirect);
#endif
  /* get number of nodes, pinums */
  bufid = pvm_recv(hosttid,STARTTYPE0);
  pvm_upkint(&nnodes,1,1);
  pvm_upkint(pinums,nnodes,1);
  pvm_upkbyte((char *)&m_msg,(int)sizeof(m_msg),1);
  worker = tidtonode(mytid);
  pvm_freebuf(bufid);
#endif

#ifdef MPI_SRC
  MPI_Recv(&m_msg,sizeof(m_msg),MPI_BYTE,hosttid,STARTTYPE0,MPI_COMM_WORLD,
	   &mpi_status);
#endif

  /* the aln structure needs some information from m_msg0.aln */
  memcpy(&aln,&m_msg.aln,sizeof(struct a_struct));

  /*
  fprintf(stderr,"d1: %d d2: %d\n",m_msg.pamd1,m_msg.pamd2);
  */

  /* get pst params */
#ifdef PVM_SRC
  bufid = pvm_recv(hosttid,STARTTYPE1);
  pvm_upkbyte((char *)&pst,(int)sizeof(pst),1);
  /* 31t nsq = pst.nsq; */
  pvm_upkbyte((char *)pascii,(int)sizeof(aascii),1);
  pvm_freebuf(bufid);
#endif
#ifdef MPI_SRC
  MPI_Recv(&pst,(int)sizeof(pst),MPI_BYTE,hosttid,STARTTYPE1,MPI_COMM_WORLD,
	   &mpi_status);

  MPI_Recv(pascii,(int)sizeof(aascii)/sizeof(int),MPI_INT,hosttid,STARTTYPE1,MPI_COMM_WORLD,
	   &mpi_status);
#endif

  aa0[5] = aa0[4] = aa0[3] = aa0[2] = aa0[1] = aa0[0];
  if (m_msg.qframe == 2) {
    if ((aa0[1]=(unsigned char *)malloc((MAXTST+2)*sizeof (char)))==NULL)
      w_abort ("Unable to allocate sequence[1] array - exiting!","");
    *aa0[1]='\0';
    aa0[1]++;
  }

  if ((aa1s=(unsigned char *)malloc((MAXTOT+1)*sizeof (char)))==NULL)
      w_abort ("Unable to allocate shuffled library sequence", "");
  *aa1s=0;
  aa1s++;

  irand(0);	/* necessary for shuffled sequences */

  /* this function allocates pam12, pam12x
     assigns pst.pam[0][0]=pam12, pst.pam[1][0] = pam12x
     and sets up the correct pst.pam[0][0][0] pointers */

  alloc_pam(m_msg.pamd1,m_msg.pamd2,&pst);

#ifdef PVM_SRC
  bufid = pvm_recv(hosttid,STARTTYPE2);
  pvm_upkint(pam12,m_msg.pamd1*m_msg.pamd2,1);
  pvm_freebuf(bufid);

  bufid = pvm_recv(hosttid,STARTTYPE3);
  pvm_upkint(pam12x,m_msg.pamd1*m_msg.pamd2,1);
  pvm_freebuf(bufid);
#endif  

#ifdef DEBUG
  if (worker==FIRSTNODE) {
    fprintf(stderr,"ext?: %d\tnsq: %d\tnsqx: %d\n",pst.ext_sq_set,pst.nsq, pst.nsqx);
    for (i=1; i<5; i++) {
      for (j=1; j <= i; j++) fprintf(stderr," %c,%c:%2d",pst.sq[i],pst.sq[j],pst.pam2[0][i][j]);
      fprintf(stderr,"\n");
    }
    for (i=pst.nsq+1; i<pst.nsq+5; i++) {
      for (j=pst.nsq+1; j <= i; j++) fprintf(stderr," %c,%c:%2d",pst.sqx[i],pst.sqx[j],pst.pam2[0][i][j]);
      fprintf(stderr,"\n");
    }

    for (i=1; i<5; i++) {
      for (j=1; j <= i; j++) fprintf(stderr," %c,%c:%2d",pst.sqx[i],pst.sqx[j],pst.pam2[1][i][j]);
      fprintf(stderr,"\n");
    }
    for (i=pst.nsq+1; i<pst.nsq+5; i++) {
      for (j=pst.nsq+1; j <= i; j++) fprintf(stderr," %c,%c:%2d",pst.sqx[i],pst.sqx[j],pst.pam2[1][i][j]);
      fprintf(stderr,"\n");
    }

  }
#endif

#ifdef MPI_SRC
  MPI_Recv(pam12,m_msg.pamd1*m_msg.pamd2,MPI_INT,hosttid,STARTTYPE2,
	   MPI_COMM_WORLD,&mpi_status);

  MPI_Recv(pam12x,m_msg.pamd1*m_msg.pamd2,MPI_INT,hosttid,STARTTYPE3,
	   MPI_COMM_WORLD,&mpi_status);
#endif

  /* Allocate space for the sequences */
  max_sql = MAXSQL/2;

  if ((seqpt=(struct sqs2 *)calloc(max_sql,sizeof(struct sqs2)))==NULL)
    w_abort("cannot allocate seqpt(sqs2)","");

  if ((n1_arr=(int *)calloc(m_msg.pbuf_siz+1,sizeof(int)))==NULL)
    w_abort("cannot allocate n1_arr","");
  if ((m_seqnm_arr=(int *)calloc(m_msg.pbuf_siz+1,sizeof(int)))==NULL)
    w_abort("cannot allocate m_seqnm_arr","");

  lcnt = 0;
  while (1) {
#ifdef PVM_SRC
    /* get the number of sequences, sequence lengths */
    bufid = pvm_recv(hosttid,STARTTYPE4);
    pvm_upkint(&seqbuf_n,1,1);	/* number of sequences */
    pvm_upkint(&seqbuf_s,1,1);	/* size of sequence buffer */
    pvm_upkint(n1_arr,seqbuf_n,1);
    pvm_upkint(m_seqnm_arr,seqbuf_n,1);
    pvm_freebuf(bufid);
#endif
#ifdef MPI_SRC
    MPI_Recv(&seqbuf_n,1,MPI_INT,hosttid,STARTTYPE4,MPI_COMM_WORLD,
	     &mpi_status);
    MPI_Recv(&seqbuf_s,1,MPI_INT,hosttid,STARTTYPE4,MPI_COMM_WORLD,
	     &mpi_status);
    MPI_Recv(n1_arr,seqbuf_n,MPI_INT,hosttid,STARTTYPE4,MPI_COMM_WORLD,
	     &mpi_status);
    MPI_Recv(m_seqnm_arr,seqbuf_n,MPI_INT,hosttid,STARTTYPE4,MPI_COMM_WORLD,
	     &mpi_status);
#endif

    if (seqbuf_n <= 0) break;
#ifdef DEBUG    
    /*
    fprintf(stderr,"[%d] seqbuf_n: %d seqbuf_s: %d\n",
	    worker,seqbuf_n,seqbuf_s);
    fprintf(stderr,"[%d] lcnt: %d n1: %d seqnm %d\n",
	    worker,0,n1_arr[0],m_seqnm_arr[0]);
    fprintf(stderr,"[%d] lcnt: %d n1: %d seqnm %d\n",
	    worker,1,n1_arr[1],m_seqnm_arr[1]);
    */
#endif

    /* allocate space for sequences */
    if ((seq_buf = (unsigned char *)calloc((size_t)seqbuf_s+1,sizeof(char)))
	==NULL) {
      w_abort("cannot allocate tmp_seq","");
    }
    seq_buf++; /* leave a '\0' at the start */

    /* get the sequence buffer */
#ifdef PVM_SRC
    bufid = pvm_recv(hosttid,STARTTYPE5);
    pvm_upkbyte(seq_buf,seqbuf_s,1);
    pvm_freebuf(bufid);
#endif  
#ifdef MPI_SRC
    MPI_Recv(seq_buf,seqbuf_s,MPI_BYTE,hosttid,STARTTYPE5,MPI_COMM_WORLD,
	     &mpi_status);
#endif

    /* now we have everything  - update the pointers */
    if (lcnt+seqbuf_n >= max_sql) {
      max_sql += max(MAXSQL/2,seqbuf_n);
      if ((seqpt=(struct sqs2 *)realloc(seqpt,max_sql*sizeof(struct sqs2)))
	  ==NULL)
	w_abort("cannot allocate seqpt(sqs2)","");
    }

    /* convert from offsets to pointers into buffer */
    ntx = 0;
    for (i=0; i<seqbuf_n; i++,lcnt++) {
      seqpt[lcnt].n1 = n1_arr[i];
      seqpt[lcnt].m_seqnm = m_seqnm_arr[i];
      seqpt[lcnt].aa1 = &seq_buf[ntx];
      ntx += n1_arr[i]+1;
    }
  }
  /* all done - lcnt has the total number of library sequences */

#ifdef DEBUG
  if (lcnt > 0)
    for (i=0; i<10; i++) {
      for (j=0; j<10; j++) libstr[j]=pst.sq[seqpt[i].aa1[j]];
      libstr[10]='\0';
      fprintf(stderr,"[%d] n1: %d seqnm: %d aa1: %s\n",
	      worker,seqpt[i].n1,seqpt[i].m_seqnm,libstr);
    }
#endif

  /* send back the number of descriptions received */

#ifdef PVM_SRC
  pvm_initsend(PvmDataRaw);
  pvm_pkint(&lcnt,1,1);
  pvm_send(hosttid,STARTTYPE0);
#endif
#ifdef MPI_SRC
/*  p4_dprintf(" have %d descriptions to send\n",lcnt); */
  MPI_Send(&lcnt,1,MPI_INT,hosttid,STARTTYPE0,MPI_COMM_WORLD);
#endif  

  /* get last set of numbers */
#ifdef PVM_SRC
  bufid = pvm_recv(hosttid,STARTTYPE0);
  pvm_upkint(last_msg_b,2,1);
  pvm_freebuf(bufid);
#endif
#ifdef MPI_SRC
  MPI_Recv(last_msg_b, 2, MPI_INT, hosttid, STARTTYPE0, MPI_COMM_WORLD,
	     &mpi_status);
#endif
  m_msg.nbr_seq = last_msg_b[0];
  qres_bufsize = last_msg_b[1];

#ifdef DEBUG
#ifdef PVM_SRC
/*
  fprintf(stderr,"[%d] have nbr_seq %d qres_bufsize %d\n",worker,
	     m_msg.nbr_seq, qres_bufsize);
*/
#endif
#ifdef MPI_SRC
  /*  p4_dprintf("[%d] have nbr_seq %d qres_bufsize %d\n",worker,
	     m_msg.nbr_seq, qres_bufsize);
  */;
#endif
#endif  
  /* If self search, receive sequence numbering data */
  if (m_msg.self) {
#ifdef PVM_SRC
    bufid = pvm_recv(hosttid,STARTTYPE1);
    pvm_upkint(&lend,1,1);
    pvm_freebuf(bufid);
#endif
#ifdef MPI_SRC
    MPI_Recv(&lend,1,MPI_INT,hosttid,STARTTYPE1,MPI_COMM_WORLD,&mpi_status);
#endif
  }
  
#if defined(FASTX) || defined(TFASTX)
  aainit(pst.tr_type, pst.debug_lib);
  pst.maxlen = m_msg.maxn;
#endif

  /* This is the loop that calls the do_work () function repeatedly */
  cur_n0 = 0;
  while (1)  {
/*
#ifdef DEBUG
#ifdef PVM_SRC
    fprintf(stderr," W: %d waiting MSEQTYPE\n",worker);
#endif
#ifdef MPI_SRC
    p4_dprintf(" W: %d waiting MSEQTYPE\n",worker);
#endif
#endif
*/
#ifdef PVM_SRC
    bufid = pvm_recv(hosttid,MSEQTYPE);
    pvm_upkbyte((char *)&qm_msg,sizeof(qm_msg),1);
#endif
#ifdef MPI_SRC
    MPI_Recv(&qm_msg,sizeof(struct mngmsg),MPI_BYTE,hosttid,MSEQTYPE,
	     MPI_COMM_WORLD,&mpi_status);
#endif
#ifdef DEBUG
/*
    fprintf(stderr,"[%d] have MSEQTYPE n0: %d s_func: %d slist: %d\n",
	    worker,qm_msg.n0,qm_msg.s_func,qm_msg.slist);
*/
#endif
    if (qm_msg.n0 > 0 && qm_msg.slist == 0) {
      if (cur_n0 > 0) {
	close_work (aa0[0], cur_n0, &pst, &f_str[0]);
	if (m_msg.qframe == 2) close_work(aa0[1], cur_n0, &pst, &f_str[1]);
      }
      cur_n0 = qm_msg.n0;
#ifdef PVM_SRC
      pvm_upkbyte((char *)aa0[0],qm_msg.n0+1,1);
#endif
#ifdef MPI_SRC
      MPI_Recv(aa0[0],qm_msg.n0+1,MPI_BYTE,hosttid,MSEQTYPE1,MPI_COMM_WORLD,
	       &mpi_status);
#endif
    }
#ifdef PVM_SRC
    pvm_freebuf(bufid);
#endif

    if (qm_msg.n0 == -1) {
/*    printf(" %d: got n0 == -1\n",worker); */
      break;
    }
    /*    p4_dprintf(" W:%d n0:%d slist:%d s_func:%d (%d)\n",worker,qm_msg.n0,qm_msg.slist,qm_msg.s_func,qres_bufsize); */

    if (qm_msg.slist > 0) {  /* list search, not library search */
      if (liblist != NULL) free(liblist);
      /* get the list of sequences */
      if ((liblist=(struct stage2_str *)
	   calloc(qm_msg.slist,sizeof(struct stage2_str)))==NULL) {
	  sprintf(errstr,"sequence list %d",qm_msg.slist);
	  w_abort (errstr, "");
	}
#ifdef PVM_SRC
      bufid = pvm_recv(hosttid,LISTTYPE);
      pvm_upkbyte((char *)liblist,qm_msg.slist*sizeof(struct stage2_str),1);
      pvm_freebuf(bufid);
#endif
#ifdef MPI_SRC
      MPI_Recv(liblist,qm_msg.slist*sizeof(struct stage2_str),MPI_BYTE,
	       hosttid,LISTTYPE,MPI_COMM_WORLD, &mpi_status);
#endif
    }

    /* Initial stuff */
    if (qm_msg.slist == 0) {
#ifdef DEBUG
/*
      fprintf(stderr,"n1: %d\t",qm_msg.n0);
      for (i=0; i<10; i++) fprintf(stderr,"%c",nt[aa0[0][i]]);
      fprintf(stderr,"\n");
*/
#endif
      init_work (aa0[0], qm_msg.n0, &pst, &f_str[0]);
      f_str[5]=f_str[4]=f_str[3]=f_str[2]=f_str[1]=f_str[0];

      if (m_msg.qframe == 2) {
	memcpy(aa0[1],aa0[0],qm_msg.n0+1);
	revcomp(aa0[1],qm_msg.n0,&pst.c_nt[0]);
	init_work (aa0[1], qm_msg.n0, &pst, &f_str[1]);
      }

#ifdef DEBUG
	fprintf(stderr,"[%d] init_work qf: %d nf: %d\n",worker,m_msg.qframe,m_msg.nframe);
#endif
    }

    bestcnt = 0;
    if (qm_msg.slist == 0) {
      for (count=0; count < lcnt; count++) {

	for (itt=m_msg.revcomp; itt<=m_msg.nitt1; itt++) {

	  rst.score[0] = rst.score[1] = rst.score[2] = 0;
	  if (m_msg.self) {
	    lsn = lend + count;
	    if ((qm_msg.seqnm > lsn) && (((qm_msg.seqnm + lsn) % 2) != 0)) {
	      do_work (aa0[itt], qm_msg.n0,seqpt[count].aa1, seqpt[count].n1,
		       itt, &pst, f_str[itt], &rst);
	    }
	    else if ((qm_msg.seqnm <= lsn) && (((qm_msg.seqnm+lsn)%2) == 0)) {
	      do_work (aa0[itt], qm_msg.n0, seqpt[count].aa1, seqpt[count].n1, 
		       itt, &pst, f_str[itt], &rst);
	    }
	    else continue;
	  }
	  else {
	    do_work (aa0[itt], qm_msg.n0, seqpt[count].aa1, seqpt[count].n1,
		     itt, &pst, f_str[itt], &rst);
	  }
#ifdef DEBUG
/*
	  if (count < 10 || (count % 200 == 199)) {
	    fprintf(stderr,"[node %d] itt:%d/%d (%d) %3d %3d %3d - %d/%d\n",
		    worker,itt,m_msg.nitt1,count,
		    rst.score[0],rst.score[1],rst.score[2],
		    seqpt[count].m_seqnm,seqpt[count].n1);
	  }
*/
#endif
	  sw_score = -1;

	  bestr[bestcnt].seqnm  = count;
	  bestr[bestcnt].m_seqnm  = seqpt[count].m_seqnm;
	  bestr[bestcnt].score[0] = rst.score[0];
	  bestr[bestcnt].score[1] = rst.score[1];
	  bestr[bestcnt].score[2] = rst.score[2];
	  bestr[bestcnt].frame = itt;
	  bestr[bestcnt].comp = rst.comp;
	  bestr[bestcnt].H = rst.H;

	  if (pst.zsflag >= 10) {
	    if (pst.zs_win > 0) 
	      wshuffle(seqpt[count].aa1, aa1s,seqpt[count].n1,pst.zs_win,&ieven);
	    else 
	      shuffle(seqpt[count].aa1, aa1s,seqpt[count].n1);

	    do_work(aa0[itt],qm_msg.n0,aa1s,seqpt[count].n1,itt, &pst, f_str[itt], &rst);
	    bestr[bestcnt].rscore = rst.score[pst.score_ix];
	  }

	  bestcnt++;
	  if (bestcnt >= qres_bufsize) {
#ifdef DEBUG
	    fprintf(stderr," worker: %d sending %d results\n",worker,qres_bufsize);
#endif
	    bestr[qres_bufsize].seqnm = bestcnt;
#ifdef PVM_SRC
	    pvm_initsend(PvmDataRaw);
	    pvm_pkbyte((char *)&bestr[0],sizeof(struct comstr)*(qres_bufsize+1),1);
	    pvm_send(hosttid,curtype);
#endif
#ifdef MPI_SRC
	  MPI_Send(bestr,sizeof(struct comstr)*(qres_bufsize+1),MPI_BYTE,
		   hosttid,curtype,MPI_COMM_WORLD);
#endif
	    bestcnt = 0;
	  }
	}
      }	/* END - for count loop */

      bestr[qres_bufsize].seqnm = bestcnt | FINISHED;
    
#ifdef PVM_SRC
#ifdef DEBUG
      fprintf(stderr,"[%d] FINISHED %d\n",worker,bestcnt);
#endif
      pvm_initsend(PvmDataRaw);
      pvm_pkbyte((char *)&bestr[0],sizeof(struct comstr)*(qres_bufsize+1),1);
      pvm_send(hosttid,curtype);
#endif
#ifdef MPI_SRC
      MPI_Send(&bestr[0],sizeof(struct comstr)*(qres_bufsize+1),MPI_BYTE,
	       hosttid,curtype,MPI_COMM_WORLD);
#endif
    }

    /* s_func == 1 means do_opt if necessary */
    else if (qm_msg.s_func==1) {  /* qm_msg.slist > 0 */
#ifdef DEBUG
	fprintf(stderr," [%d] starting s_func:1 slist: %d\n",
		worker,qm_msg.slist);
#endif
      for (count=0; count < qm_msg.slist; count++) {
	rst.score[0] = rst.score[1] = rst.score[2] = 0;
	itt = liblist[count].frame;
	seqnm = liblist[count].seqnm;

	bestr2[bestcnt].seqnm  = seqnm;
	bestr2[bestcnt].m_seqnm = seqpt[seqnm].m_seqnm;
	if (m_msg.stages > 1) {
	  do_opt (aa0[itt], qm_msg.n0, seqpt[seqnm].aa1,
		  seqpt[seqnm].n1, itt,
		  &pst, f_str[itt], &rst);

	  bestr2[bestcnt].score[0] = rst.score[0];
	  bestr2[bestcnt].score[1] = rst.score[1];
	  bestr2[bestcnt].score[2] = rst.score[2];
	}

	if (m_msg.markx == 9) {
#ifdef DEBUG
	  fprintf(stderr," [%d] starting do_walign seqnm: %d n1: %d\n",
		  worker,seqnm,seqpt[seqnm].n1);
#endif
	  aln_dp = &bestr2[bestcnt].aln_d;
	  memcpy(aln_dp, &aln,sizeof(struct a_struct));

	  sw_score=
	    do_walign(aa0[itt], qm_msg.n0,seqpt[seqnm].aa1,
		      seqpt[seqnm].n1, itt,
		      &pst, f_str[itt], &ares, &nres, aln_dp);

#ifdef DEBUG
	  fprintf(stderr," [%d] starting calc_id sw_score: %d\n",
		  worker,sw_score);
#endif
	  lc=calc_id(aa0[itt],qm_msg.n0, seqpt[seqnm].aa1,
		     seqpt[seqnm].n1,
		     ares,nres,aln_dp,pst,f_str[itt]);

	  nident = aln_dp->nident;
	  aln_dp->a_len = lc;

	  if (lc > 0) percent = (100.0*(float)nident)/(float)lc;
	  else percent = 0.0;

	  ngap = aln_dp->ngap_q + aln_dp->ngap_l;
	  if (lc-ngap > 0) gpercent = (100.0*(float)nident)/(float)(lc-ngap);
	  else gpercent = 0.0;

	  bestr2[bestcnt].swscore = sw_score;
	  /*
	  bestr2[bestcnt].a_len = lc;
	  bestr2[bestcnt].min0 = aln.min0;
	  bestr2[bestcnt].max0 = aln.max0;
	  bestr2[bestcnt].min1 = aln.min1;
	  bestr2[bestcnt].max1 = aln.max1;
	  bestr2[bestcnt].ngap_q = aln.ngap_q;
	  bestr2[bestcnt].ngap_l = aln.ngap_l;
	  */
	  bestr2[bestcnt].percent = percent;
	  bestr2[bestcnt].gpercent = gpercent;
	}
	bestcnt++;

	if (bestcnt >= BFR2) {
	  bestr2[BFR2].seqnm = bestcnt;
#ifdef PVM_SRC
	  pvm_initsend(PvmDataRaw);
	  pvm_pkbyte((char *)&bestr2[0],sizeof(struct comstr2)*(BFR2+1),1);
	  pvm_send(hosttid,LISTRTYPE);
#endif
#ifdef MPI_SRC
	  MPI_Send(&bestr2[0],sizeof(struct comstr2)*(BFR2+1),MPI_BYTE,
		   hosttid,LISTRTYPE,MPI_COMM_WORLD);
#endif
	  bestcnt = 0;
	}
      }	/* END - for count loop */

      bestr2[BFR2].seqnm = bestcnt | FINISHED;
#ifdef PVM_SRC
      pvm_initsend(PvmDataRaw);
      pvm_pkbyte((char *)&bestr2[0],sizeof(struct comstr2)*(BFR2+1),1);
      pvm_send(hosttid,LISTRTYPE);
#ifdef DEBUG
      fprintf(stderr,"[%d] FINISHED %d\n",worker,bestcnt);
#endif
#endif
#ifdef MPI_SRC
      MPI_Send(&bestr2[0],sizeof(struct comstr2)*(BFR2+1),MPI_BYTE,
	       hosttid,LISTRTYPE,MPI_COMM_WORLD);
#endif
    }
    /* get alignments */
    else if (qm_msg.s_func==2) {
      for (count=0; count < qm_msg.slist; count++) {
	itt = liblist[count].frame;
	seqnm = liblist[count].seqnm;
/*
	fprintf(stderr,"worker: %d; %s, frame: %d\n",worker,qm_msg.libstr,itt);
*/
	sw_score =
	  do_walign (aa0[itt], qm_msg.n0,seqpt[seqnm].aa1,
		     seqpt[seqnm].n1, itt,
		     &pst, f_str[itt], &ares, &nres, &aln);

	if (aln.showall==1)
	  maxc = nres + max(aln.min0,aln.min1)+
	    max((qm_msg.n0-aln.max0),
		(seqpt[seqnm].n1-aln.max1))+4;
	else  maxc = nres + 4*aln.llen+4;
	initseq(&seqc0, &seqc1, maxc);

	nc=calcons(aa0[itt],qm_msg.n0, seqpt[seqnm].aa1, seqpt[seqnm].n1,
		   ares,nres,&lc,&aln,pst,seqc0,seqc1,f_str[itt]);

	nident = aln.nident;
	percent = (100.0*(float)nident)/(float)lc;
	ngap = aln.ngap_q+aln.ngap_l;
	if (lc-ngap > 0) {
	  gpercent = (100.0*(float)nident)/(float)(lc-ngap);
	}

#ifdef PVM_SRC
	pvm_initsend(PvmDataRaw);
	pvm_pkint(&nc,1,1);
	pvm_pkint(&lc,1,1);
	pvm_pkfloat(&percent,1,1);
	pvm_pkfloat(&gpercent,1,1);
/* 	pvm_pkint(&seqpt[liblist[count]].n1,1,1); */
	pvm_pkint(&sw_score,1,1);
	pvm_pkbyte((char *)&aln,sizeof(struct a_struct),1);
	pvm_send(hosttid,ALN1TYPE);
#ifdef DEBUG
	fprintf(stderr,"[%d] ALN1TYPE sent: %d\n",worker,qm_msg.n0);
#endif
	pvm_initsend(PvmDataRaw);
	pvm_pkbyte(seqc0,nc,1);
	pvm_pkbyte(seqc1,nc,1);
	pvm_send(hosttid,ALN2TYPE);
#endif
#ifdef MPI_SRC
	last_msg_b[0]=nc;
	last_msg_b[1]=lc;
	last_msg_b[2]=sw_score;
	MPI_Send(last_msg_b,3,MPI_INT,hosttid,ALN1TYPE,MPI_COMM_WORLD);
	MPI_Send(&percent,1,MPI_FLOAT,hosttid,ALN2TYPE,MPI_COMM_WORLD);
	MPI_Send(&gpercent,1,MPI_FLOAT,hosttid,ALN2TYPE,MPI_COMM_WORLD);

/*	p4_dprintf("[%d] sending aln\n",worker); */
	MPI_Send(&aln,sizeof(struct a_struct),MPI_BYTE,hosttid,
		 ALN3TYPE,MPI_COMM_WORLD);

	MPI_Send(seqc0,nc,MPI_BYTE,hosttid,ALN2TYPE,MPI_COMM_WORLD);
	MPI_Send(seqc1,nc,MPI_BYTE,hosttid,ALN3TYPE,MPI_COMM_WORLD);
#endif
	freeseq(&seqc0,&seqc1);
      }	
    }
    
/* send back parameter settings */
    if (worker==FIRSTWORK && qm_msg.slist==0) {
      get_param(&pst, gstring2,gstring3);
#ifdef PVM_SRC
      pvm_initsend(PvmDataRaw);
      pvm_pkbyte(gstring2,sizeof(gstring2),1);
      pvm_pkbyte(gstring3,sizeof(gstring3),1);
      pvm_send(hosttid,PARAMTYPE);
#endif
#ifdef MPI_SRC
      MPI_Send(gstring2,sizeof(gstring2),MPI_BYTE,
	       hosttid,PARAMTYPE,MPI_COMM_WORLD);
#endif
    }
    
    if (qm_msg.slist==0) {
      if (curtype == ONETYPE) curtype = TWOTYPE;
      else curtype = ONETYPE;
    }
  }	    /* END - while (1) loop */
#ifdef PVM_SRC
  pvm_exit();
#endif
}

#ifdef PVM_SRC
int tidtonode(tid)
     int tid;
{
  int i;
  for (i=FIRSTNODE; i< nnodes; i++) if (tid==pinums[i]) return i;
  fprintf(stderr," cannot find tid %d\n",tid);
  return -1;
}
#endif
